package com.motu.vertx.module.utility.logsdk;

import com.motu.vertx.module.utility.base.BaseServerParam;
import com.motu.vertx.module.utility.base.ServerConfig;
import com.motu.vertx.module.utility.kafka.KafkaManager;
import com.motu.vertx.module.utility.toolset.Tool;
import io.vertx.core.Vertx;
import io.vertx.core.json.JsonObject;
import io.vertx.kafka.client.producer.KafkaProducer;
import io.vertx.kafka.client.producer.KafkaProducerRecord;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

public class KafkaBiManager {
	public static final Logger logger = LoggerFactory.getLogger(KafkaBiManager.class);
    public static KafkaProducer<String, String> producer;
    private static boolean isOnlineServer;
    private static String gameId;
	public static void init(Vertx vertx, JsonObject appConfig){
		String kafkaAddress = appConfig.getString("bi-kafka-address");
		String gid = appConfig.getString("bi-gameid");
		if (Tool.isEmpty(kafkaAddress) || Tool.isEmpty(gid)){
			return;
		}

		Properties config = new Properties();
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaAddress);
        config.put("key.serializer", StringSerializer.class);
        config.put("value.serializer", StringSerializer.class);
        config.put("acks", "1");
		// 优化kafka生产者配置
		KafkaManager.upgradeProducerConfig(config, appConfig);

		// 创建一个Kafka Producer
        producer = KafkaProducer.create(vertx, config);
        
        isOnlineServer = ServerConfig.isOnlineServer();
        gameId = gid;
		logger.info("bi kafka create success address:{} gameId:{} isOnlineServer:{}}", kafkaAddress, gameId, isOnlineServer);
    }
    private static int getPartitionId(String logType) {
    	if(logType.equalsIgnoreCase(NewbiRequestInfo.TYPE_REGISTER)) {
    		return 0;
    	}else if(logType.equalsIgnoreCase(NewbiRequestInfo.TYPE_LOGIN)) {
    		return 1;
    	}else if(logType.equalsIgnoreCase(NewbiRequestInfo.TYPE_PAYMENT)) {
    		return 2;
    	}else if(logType.equalsIgnoreCase(NewbiRequestInfo.TYPE_EVENT)) {
    		return 3;
    	}else if(logType.equalsIgnoreCase(NewbiRequestInfo.TYPE_INSTALL)) {
    		return 4;
    	}else if(logType.equalsIgnoreCase(NewbiRequestInfo.TYPE_STARTUP)) {
    		return 5;
    	}else if(logType.equalsIgnoreCase(NewbiRequestInfo.TYPE_MONITOR)) {
    		return 6;
    	}else {
    		logger.error("getPartitionId error logType:"+logType);
    	}
    	return 0;
    }
    // 日志发送到消息队列
    private static void sendLog(String logType, JsonObject params) {
//    	System.out.println("KafkaBiManager1 send logType:"+logType+" params:"+params);
    	
//        KafkaProducerRecord<String, JsonObject> record = KafkaProducerRecord.create(gameId, logType, params,getPartitionId(logType));
        KafkaProducerRecord<String, String> record = KafkaProducerRecord.create(gameId,params.toString());
        producer.send(record, done -> {
            if (done.failed()) {
               logger.error("pushDBQueue fail cause = " + Tool.getException(done.cause()));
            }else {
//            	logger.info("KafkaBiManager send logType:"+logType+" params:"+params);
//            	 RecordMetadata recordMetadata = done.result();
//            	logger.info("pushDBQueue Message " + record.value().toString() + " written on topic=" + recordMetadata.getTopic() +
//                        ", partition=" + recordMetadata.getPartition() +
//                        ", offset=" + recordMetadata.getOffset());
            }
        });
    }
    
    public static void addLog(NewbiRequestInfo info,int logTime) {
    	if(!isOnlineServer) {
    		return;
    	}
		if (BaseServerParam.SEND_BI_DATA == 0){
			return;
		}
    	JsonObject jo = new JsonObject();
    	String who = ""+info.userId;
		JsonObject context = new JsonObject();
		
		context.put("_ip",info.ip);
		context.put("_deviceid",info.deviceid);
		if(info.ryos.equalsIgnoreCase("ios")) {
			context.put("_idfa",info.idfa);
			context.put("_idfv",info.idfv);
		}else if(info.ryos.equalsIgnoreCase("android")) {
			context.put("_imei",info.imei);
			context.put("_androidid",info.androidid);
			context.put("_oaid",info.oaid);
		}
		if(info.logType.equalsIgnoreCase(NewbiRequestInfo.TYPE_PAYMENT)) {

			context.put("_transactionid", info.orderId);
			context.put("_paymenttype", "unknown");
			context.put("_currencytype", BaseServerParam.LOG_CURRENCY_TYPE);
			context.put("_currencyamount", info.money);
			context.put("_rydevicetype", ""+info.deviceType);

		}else if(info.logType.equalsIgnoreCase(NewbiRequestInfo.TYPE_REGISTER )) {
			context.put("_rydevicetype", ""+info.deviceType);
		}else if(info.logType.equalsIgnoreCase(NewbiRequestInfo.TYPE_EVENT )) {
			context.put("_eventparam1", info.eventParam1);
			context.put("_eventparam2", info.eventParam2);
			jo.put("what",info.eventName);
		}else if(info.logType.equalsIgnoreCase(NewbiRequestInfo.TYPE_MONITOR )) {
			context.put("_monitortype", info.monitorType);
			context.put("_monitorparam", info.monitorParam);
		}
		jo.put("appid", gameId);
		jo.put("who",who);
		jo.put("accountid",info.accountId);
		jo.put("platform",info.platform);
		jo.put("channel",info.channel);
		jo.put("context", context);
		jo.put("logtype",info.logType);
		if(logTime > 0) {//重发的数据，日志时间不能用当前时间，否则影响数据统计
			jo.put("logtime",logTime);
		}
		sendLog(info.logType, jo);
    }
    
    public static void addWordMonitorLog(String accountId,String regDeviceId, long userId,JsonObject gobj,int monitorType,String monitorParam) {
    	if(!isOnlineServer) {
    		return;
    	}
    	NewbiRequestInfo biinfo = new NewbiRequestInfo(NewbiRequestInfo.TYPE_MONITOR, accountId,regDeviceId, 0, 0, userId, gobj,monitorType,monitorParam);
		KafkaBiManager.addLog(biinfo, 0);
    }

}
